package zzjson.rabbitmq.delay;

import com.rabbitmq.client.*;
import zzjson.rabbitmq.BaseRabbitMq;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * <p>****************************************************************************</p>
 * <p><b>Copyright © 2010-2019 shuncom team All Rights Reserved<b></p>
 * <ul style="margin:15px;">
 * <li>Description : com.zzjson.rabbitmq.message</li>
 * <li>Version     : 1.0.0</li>
 * <li>Creation    : 2019年04月25日</li>
 * <li>@author     : zzy0_0</li>
 * </ul>
 * <p>****************************************************************************</p>
 */
public class DelayPluginConsumer {
	public static void main(String[] args) throws IOException {
		Connection conn = BaseRabbitMq.getConn();
		Channel channel = conn.createChannel();

		// 声明x-delayed-type类型的exchange
		Map<String, Object> argss = new HashMap<String, Object>();
		argss.put("x-delayed-type", "direct");
		channel.exchangeDeclare("DELAY_EXCHANGE", "x-delayed-message", false,
				false, argss);

		// 声明队列
		channel.queueDeclare("DELAY_QUEUE", false, false, false, null);

		// 绑定交换机与队列
		channel.queueBind("DELAY_QUEUE", "DELAY_EXCHANGE", "DELAY_KEY");

		// 创建消费者
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
									   byte[] body) throws IOException {
				String msg = new String(body, StandardCharsets.UTF_8);
				SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
				System.out.println("Received message : '" + msg + "', at " + sf.format(new Date()));
			}
		};

		// 开始获取消息
		// String queue, boolean autoAck, Consumer callback
		channel.basicConsume("DELAY_QUEUE", true, consumer);

		BaseRabbitMq.close(channel, conn);

	}
}